package com.example.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class DemoApplication {

	public static void main(String[] args) throws Exception {
		SpringApplication.run(DemoApplication.class, args);
//		//创建执行环境
//		ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
//		//从文件中读取文件
//		DataSource<String> stringDataStreamSource = executionEnvironment.readTextFile("demo.txt");
//		//输出 生成二元组数据
//		FlatMapOperator<String, Tuple2<String, Long>> returns = stringDataStreamSource.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
//			//对每一行文本进行拆分
//			String[] words = line.split(" ");
//			for (String word : words) {
//				out.collect(Tuple2.of(word, 1L));
//			}
//		}).returns(Types.TUPLE(Types.STRING, Types.LONG));
//		//按照 word进行分组
//		UnsortedGrouping<Tuple2<String, Long>> tuple2UnsortedGrouping = returns.groupBy(0);
//		//累加
//		AggregateOperator<Tuple2<String, Long>> sum = tuple2UnsortedGrouping.sum(1);
//		//结果
//		sum.print();
		System.out.println("项目启动后执行");
//		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//		env.setParallelism(1);
////		DataStream<Tuple2<String, Integer>> dataStreaming = env
////				.socketTextStream("localhost", 9998)
////				.flatMap(new Splitter())
////				.keyBy(0)
////				.timeWindow(Time.seconds(5))
////				.sum(1);
////		dataStreaming.print();
////		// execute program
////		env.execute("Flink Streaming Java API Skeleton");
//		DataStreamSource<String> localhost = env.socketTextStream("localhost", 9998);
//		DataStream<Tuple2<String, Integer>> sum =
//				localhost.flatMap(new Splitter()).keyBy(0).sum(1);
//		sum.print();
//		env.execute();
	}

	public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
		@Override
		public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
			for(String word : sentence.split(" ")){
				out.collect(new Tuple2<String, Integer>(word, 1));
			}
		}
	}
}
